#include <iostream>
#include <thread>
#include <mutex>
#include <vector>
#include <queue>
#include <functional>
#include <condition_variable>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fstream>
#include <unistd.h>
#include <future>


class ThreadPool {
private:
    std::vector<std::thread> threads;         // 线程对象容器
    std::queue<std::function<void()>> tasks;  // 任务队列
    
    std::mutex queue_mutex;                   // 对queue_mutex进行加锁和解锁，确保同一时刻只有一个线程能够修改任务队列
    std::condition_variable condition;        // 用于实现线程的条件等待和通知机制，用来唤醒等待状态的线程
    bool stop;                                // 用于指示线程池是否应该停止运行。初始化为 false，表示线程池开始时不应该停止 
    
public:
    ThreadPool(size_t num_threads) : stop(false) {
        for (size_t i = 0; i < num_threads; i++) {
            threads.emplace_back([this] { // 定义每个线程对象的行为

                while (true) {
                    // 存储待执行任务的 std::function
                    std::function<void()> task;
                    
                    // 从任务队列中取任务
                    {
                        std::unique_lock<std::mutex> lock(queue_mutex);

                        // 一个等待循环，线程在这里等待条件满足（即 stop=true 或者任务队列不为空）
                        // lock 作为参数传递给 wait 函数的互斥锁，会在等待过程中被释放，允许其他线程获取这个锁
                        condition.wait(lock, [this] { return stop || !tasks.empty(); });
                        
                        // 再次检查，若 stop=true 且任务队列为空时能退出循环，而不是等待条件再次满足
                        if (stop && tasks.empty()) return;
                        
                        // 从队头取出任务
                        task = std::move(tasks.front());
                        tasks.pop();
                    }
                    // 线程执行任务
                    task();
                }
            });
        }
    }
    
    ~ThreadPool() {
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            // 表示线程池应该停止运行
            stop = true;
        }
        // 调用 condition（条件变量）的 notify_all 函数，用于唤醒所有正在等待的线程
        // 这是为了通知等待在条件变量上的线程，以便它们能够检查 stop 标志并在必要时退出循环
        condition.notify_all();
        
        // 对每个工作线程调用 join 函数，等待线程的结束
        // join 会阻塞调用线程，直到被调用的线程执行完成。这确保了在销毁 ThreadPool 对象之前，所有工作线程都已经执行完毕
        for (std::thread& thread : threads) {
            thread.join();
        }
    }
    
    template <class F, class... Args> // 模板函数，可以接受任意类型的可调用对象（函数、Lambda 表达式等）和其参数
    void enqueue(F&& f, Args&&... args) { // f可以是lambda（右值），因为有完美转发
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            tasks.emplace([=] () { std::forward<F>(f)(std::forward<Args>(args)...); }); // 完美转发函数f和参数
        }
        condition.notify_one(); // 条件变量，通知线程可以处理了
    }
};

/*
    加入：
    - 双任务队列，生产者消费者不会竞争同一个队列
    - 给stop加了锁，让锁的粒度更细
*/
class ThreadPool_DoubleQueue {
private:
    std::vector<std::thread> threads;           // 线程对象容器
    std::queue<std::function<void()>> tasks1;   // 第一个任务队列
    std::queue<std::function<void()>> tasks2;   // 第二个任务队列
    
    std::mutex queue_mutex;                     // 任务队列锁
    std::mutex stop_mutex;                      // 停止标志锁
    std::condition_variable condition;          // 条件变量，用于唤醒线程
    bool stop;                                  // 用于指示线程池是否应该停止运行

    // 任务交换函数：将 tasks1 和 tasks2 交换
    void swap_tasks() {
        std::swap(tasks1, tasks2);              // 在锁住的情况下交换任务队列
        // PS：不会导致拷贝带来的代价，底层是直接交换指针
    }

public:
    ThreadPool_DoubleQueue(size_t num_threads) : stop(false) {
        for (size_t i = 0; i < num_threads; i++) {
            threads.emplace_back([this] {
                while (true) {
                    std::function<void()> task;

                    {
                        std::unique_lock<std::mutex> lock(queue_mutex);

                        // 只有在任务队列为空且线程池未停止时，才进入等待
                        condition.wait(lock, [this] { 
                            std::lock_guard<std::mutex> stop_lock(stop_mutex); // 检查停止标志时单独锁住
                            return stop || !tasks1.empty() || !tasks2.empty(); 
                        });
                        
                        // 线程池停止且任务队列为空时，退出
                        {
                            std::lock_guard<std::mutex> stop_lock(stop_mutex);
                            if (stop && tasks1.empty() && tasks2.empty()) return;
                        }

                        // 如果任务1为空，尝试交换
                        if (tasks1.empty()) {
                            swap_tasks();
                        }

                        // 从任务队列2中取任务
                        if (!tasks1.empty()) {
                            task = std::move(tasks1.front());
                            tasks1.pop();
                        }
                    }
                    if (task) task();  // 执行任务
                }
            });
        }
    }
    
    ~ThreadPool_DoubleQueue() {
        {
            std::lock_guard<std::mutex> lock(stop_mutex);
            stop = true;
        }

        condition.notify_all();  // 唤醒所有等待的线程
        for (std::thread& thread : threads) {
            thread.join();  // 等待所有线程结束
        }
    }

    template <class F, class... Args>
    void enqueue(F&& f, Args&&... args) {
        {
            std::lock_guard<std::mutex> lock(queue_mutex);
            // 生产者，放任务到生产队列
            tasks2.emplace([=] { std::forward<F>(f)(std::forward<Args>(args)...); });
        }
        condition.notify_one();  // 唤醒一个等待的线程
    }
};


/*
    加入：
    - 动态创建线程，设置线程上限
    - 空闲时销毁进程，节省服务器资源
    - 最小的线程保留，避免没有任务时所有线程都销毁了
*/
class ThreadPool {
private:
    std::vector<std::thread> threads;           // 线程对象容器
    std::queue<std::function<void()>> tasks1;   // 第一个任务队列
    std::queue<std::function<void()>> tasks2;   // 第二个任务队列
    
    std::mutex queue_mutex;                     // 任务队列锁
    std::mutex stop_mutex;                      // 停止标志锁
    std::condition_variable condition;          // 条件变量，用于唤醒线程
    bool stop;                                  // 用于指示线程池是否应该停止运行

    const size_t min_threads;                   // 保留的最小线程数
    const size_t max_threads;                   // 线程池的最大线程数
    const size_t idle_timeout_ms;               // 线程的闲置超时时间

    // 任务交换函数：将 tasks1 和 tasks2 交换
    void swap_tasks() {
        std::swap(tasks1, tasks2);              // 在锁住的情况下交换任务队列
    }

public:
    ThreadPool(size_t num_threads, size_t max_threads, size_t idle_timeout_ms, size_t min_threads)
        : stop(false), max_threads(max_threads), idle_timeout_ms(idle_timeout_ms), min_threads(min_threads) {
        for (size_t i = 0; i < num_threads; i++) {
            create_thread();
        }
    }
    
    ~ThreadPool() {
        {
            std::lock_guard<std::mutex> lock(stop_mutex);
            stop = true;
        }

        condition.notify_all();  // 唤醒所有等待的线程
        for (std::thread& thread : threads) {
            thread.join();  // 等待所有线程结束
        }
    }

    // 动态创建线程的函数
    void create_thread() {
        threads.emplace_back([this] {
            while (true) {
                std::function<void()> task;

                {
                    std::unique_lock<std::mutex> lock(queue_mutex);
                    
                    // 使用带超时的等待
                    if (!condition.wait_for(lock, std::chrono::milliseconds(idle_timeout_ms), [this] {
                        std::lock_guard<std::mutex> stop_lock(stop_mutex);
                        return stop || !tasks1.empty() || !tasks2.empty();
                    })) {
                        std::lock_guard<std::mutex> stop_lock(stop_mutex);
                        // 检查是否超时，且线程数量是否大于最小线程数
                        if (threads.size() > min_threads) {
                            // 超时且可以销毁线程，退出线程
                            return;
                        }
                    }
                    
                    // 再次检查停止标志
                    {
                        std::lock_guard<std::mutex> stop_lock(stop_mutex);
                        if (stop && tasks1.empty() && tasks2.empty()) return;
                    }

                    // 如果任务1为空，尝试交换
                    if (tasks1.empty()) {
                        swap_tasks();
                    }

                    // 从任务队列1中取任务
                    if (!tasks1.empty()) {
                        task = std::move(tasks1.front());
                        tasks1.pop();
                    }
                }

                if (task) task();  // 执行任务
            }
        });
    }

    // 动态管理线程：当任务过多时创建新线程
    void manage_threads() {
        std::lock_guard<std::mutex> lock(queue_mutex);
        if (tasks2.size() > threads.size() && threads.size() < max_threads) {
            create_thread();  // 创建新的线程
        }
    }

    template <class F, class... Args>
    void enqueue(F&& f, Args&&... args) {
        {
            std::lock_guard<std::mutex> lock(queue_mutex);
            tasks2.emplace([=] { std::forward<F>(f)(std::forward<Args>(args)...); });
        }
        condition.notify_one();  // 唤醒一个等待的线程
        manage_threads();        // 动态管理线程
    }
};


// 每个线程响应客户端的 handle_client 函数
void handle_client(int client_socket) {
    std::ifstream html_file("index.html");
    std::string html_content((std::istreambuf_iterator<char>(html_file)), std::istreambuf_iterator<char>());
    
    // 构建HTTP响应头
    std::string response = "HTTP/1.1 200 OK\r\n";
    response += "Content-Length: " + std::to_string(html_content.size()) + "\r\n";
    response += "Content-Type: text/html\r\n\r\n";
    
    // 发送HTTP响应头和HTML内容给客户端
    send(client_socket, response.c_str(), response.size(), 0);
    send(client_socket, html_content.c_str(), html_content.size(), 0);
    
    // 测试，添加延时，模拟服务器处理时间
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
    
    // 关闭连接
    close(client_socket);
}


int main() {
    const int PORT = 8089;
    const int BACKLOG = 5000; // 最大连接数
    const size_t THREAD_POOL_SIZE = 1000;  // 设置线程池大小
    
    // 全局互斥锁，用于保护对 std::cout 的访问
    std::mutex cout_mutex;
    
    // 服务器socket设置和绑定的部分
    // 创建服务器socket
    int server_socket = socket(AF_INET, SOCK_STREAM, 0);
    if (server_socket == -1) {
        perror("Error creating server socket");
        return 1;
    }
    
    // 设置服务器地址结构 sockaddr_in
    sockaddr_in server_address{};
    server_address.sin_family = AF_INET;
    server_address.sin_addr.s_addr = INADDR_ANY;  // 监听所有网络接口
    server_address.sin_port = htons(PORT);  // 设置监听的端口
    
    // 绑定服务器 socket 到地址结构
    if (bind(server_socket, (struct sockaddr*)&server_address, sizeof(server_address)) == -1) {
        perror("Error binding server socket");
        close(server_socket);
        return 1;
    }
    
    if (listen(server_socket, BACKLOG) == -1) {
        perror("Error listening on server socket");
        close(server_socket);
        return 1;
    }
    
    
    ThreadPool thread_pool(THREAD_POOL_SIZE);  // 创建线程池
    
    
    while (true) {
        
        sockaddr_in client_address{}; // 结构体 sockaddr_in，用于存储客户端的地址信息
        socklen_t client_address_size = sizeof(client_address); // 获取 sockaddr_in 结构体的大小，用于传递给 accept 函数
        
        int client_socket = accept(server_socket, (struct sockaddr*)&client_address, &client_address_size); // 接受客户端的连接请求
        if (client_socket == -1) {
            perror("Error accepting client connection");
            continue;
        }
        
        // 用于锁住 cout_mutex。这是为了确保输出到标准输出时是线程安全的
        {
            std::lock_guard<std::mutex> lock(cout_mutex);
            std::cout << "Connection from " << inet_ntoa(client_address.sin_addr) << ":" << ntohs(client_address.sin_port) << std::endl;
        }
        
        // 将处理客户端请求的任务添加到线程池中
        thread_pool.enqueue(handle_client, client_socket);
    }
    
    close(server_socket);
    
    return 0;
}



